﻿using System;
using System.Collections.Concurrent;
using System.Collections.Generic;
using System.IO;
using System.Text;
using System.Threading;
using System.Threading.Tasks;
using Microsoft.Extensions.Logging;
using MQTTnet;
using MQTTnet.Client;　　　　　　//客户端需要用到
using MQTTnet.Client.Options;  //具体连接时需要用到的属性,ID的名称,要连接Server的名称,接入时用到的账号和密码,掉线时是否重新清除原有名称,还有许多...
using MQTTnet.Packets;　　　　//这个没用上
using MQTTnet.Protocol;　　  //这个也没用上
using MQTTnet.Client.Receiving;　　　　//接收
using MQTTnet.Client.Disconnecting;　　//断线
using MQTTnet.Client.Connecting;    //连接
using Newtonsoft.Json;
using Model;
using Model.Config;
using Model.Entities_EF;

namespace Service
{
    public class Mqtt
    {
        private readonly ILogger<Mqtt> _logger;
        private readonly ICPrkmonitorContext _icPrkmonitorContext;
        private readonly MqttSettingsModel _mqttSettingsModel;

        public Mqtt(ILogger<Mqtt> logger, ICPrkmonitorContext icPrkmonitorContext, MqttSettingsModel mqttSettingsModel)
        {
            _logger = logger;
            _icPrkmonitorContext = icPrkmonitorContext;
            _mqttSettingsModel = mqttSettingsModel;
        }
        private IMqttClient _mqttClient = null;

        public async void StartAsync()
        {
            try
            {
                var options = new MqttClientOptions() { ClientId = Guid.NewGuid().ToString() };
                options.ChannelOptions = new MqttClientTcpOptions()
                {
                    Server = _mqttSettingsModel.Server,
                    Port = Convert.ToInt32(_mqttSettingsModel.Port)
                };
                options.Credentials = new MqttClientCredentials()
                {
                    Username = _mqttSettingsModel.Username,
                    Password = Encoding.Default.GetBytes(_mqttSettingsModel.Password)
                };

                options.CleanSession = true;
                options.KeepAlivePeriod = TimeSpan.FromSeconds(10);
                options.KeepAliveSendInterval = TimeSpan.FromSeconds(10);
                if (null != _mqttClient)
                {
                    await _mqttClient.DisconnectAsync();
                    _mqttClient = null;
                }
                _mqttClient = new MqttFactory().CreateMqttClient();

                MqttClientStartAsync(options);
                _mqttClient.UseDisconnectedHandler(async e =>
                {
                    Console.WriteLine("### DISCONNECTED FROM SERVER ###");
                    await Task.Delay(TimeSpan.FromSeconds(2));

                    try
                    {
                        MqttClientStartAsync(options);
                        Console.WriteLine("### restart sucess ###");
                    }
                    catch
                    {
                        Console.WriteLine("### RECONNECTING FAILED ###");
                    }
                });
            }
            catch (Exception exception)
            {
             var Messages =   exception.Message;
            }
        }

        public async void MqttClientStartAsync(MqttClientOptions options)
        {
            await _mqttClient.ConnectAsync(options);
            await _mqttClient.SubscribeAsync("JH/Data");
            _mqttClient.ApplicationMessageReceivedHandler = new MqttApplicationMessageReceivedHandlerDelegate(OnSubscriberMessageReceived);
        }
        /// <summary>
        /// 反序列化为对象
        /// </summary>
        /// <typeparam name="T"></typeparam>
        /// <param name="json"></param>
        /// <returns></returns>
        public static T Deserialize<T>(string json)
        {
            T obj = Activator.CreateInstance<T>();
            using (MemoryStream ms = new MemoryStream(Encoding.Unicode.GetBytes(json)))
            {
                System.Runtime.Serialization.Json.DataContractJsonSerializer serializer = new System.Runtime.Serialization.Json.DataContractJsonSerializer(obj.GetType());
                obj = (T)serializer.ReadObject(ms);
                ms.Close();
                return obj;
            }
        }
        private async void OnSubscriberMessageReceived(MqttApplicationMessageReceivedEventArgs x)
        {
            var payloadString = x.ApplicationMessage.ConvertPayloadToString();
            var realTimeData = Deserialize<RealTimeData>(payloadString);
           await RsServer_OnReceiveRealtimeDataAsync(realTimeData);
        }
        private async Task RsServer_OnReceiveRealtimeDataAsync(RealTimeData data)
        {
            try
            {
                _logger.LogWarning($"时间{DateTime.Now}收到{data.NodeList.Count}条");
                foreach (NodeData ndata in data.NodeList)//遍历节点数据。数据包括网络设备的数据以及各个节点数据。温湿度数据存放在节点数据中
                {

                    var tbhistory = new Tbhistory()
                    {
                        Id = string.Join("", Guid.NewGuid().ToString().Split('-')),
                        //DeviceName = "温湿度设备",
                        DeviceAddr = data.DeviceID,
                        NodeId = ndata.NodeID,
                        Hum = ndata.Hum,
                        Tem = ndata.Tem,
                        CoordinateType = Convert.ToByte(ndata.CoordinateType),
                        Lng = data.Lng,
                        Lat = data.Lat,
                        RecordTime = DateTime.Now,
                        //IsAlarmData = Convert.ToByte(true),
                        //Source = Convert.ToByte(true)
                    };

                    //if (_tbhistoryQueue.Count < queuecount)
                    //{
                    //    _tbhistoryQueue.Enqueue(tbhistory);
                    //    continue;
                    //    //  return $"入队列时间:{DateTime.Now} 成功添加队列共" + _tbhistoryQueue.Count + "条数据";
                    //}
                    var result =await TbhistoryToQueueAsync(tbhistory);
                    if (!string.IsNullOrWhiteSpace(result))
                    {
                        _logger.LogInformation(result);
                    }
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e.Message);
            }

        }

        private readonly int queuecount = 100;
        private readonly ConcurrentQueue<Tbhistory> _tbhistoryQueue = new ConcurrentQueue<Tbhistory>();
        public async Task<string> TbhistoryToQueueAsync(Tbhistory tbhistory)
        {
            lock (_tbhistoryQueue)
            {
                if (!(_tbhistoryQueue.Count >= queuecount))
                {
                    _tbhistoryQueue.Enqueue(tbhistory);
                    return null;
                }
                return  Data_InstallAsync(_tbhistoryQueue).GetAwaiter().GetResult();
            }

        }
        /// <summary>
        /// 数据入库
        /// </summary>
        /// <param name="tbhistory"></param>
        /// <returns></returns>
        public async Task<string>  Data_InstallAsync(ConcurrentQueue<Tbhistory> _tbhistoryQueue)
        {
            try
            {
                string str = $"数据入库 方法收到{_tbhistoryQueue.Count} 条 \r\n";
                List<Tbhistory> listHistory = new List<Tbhistory>();
                lock (listHistory)
                {
                    lock (_tbhistoryQueue)
                    {
                        while (_tbhistoryQueue.Count > 0)
                        {
                            _tbhistoryQueue.TryDequeue(out Tbhistory dequeueTbhistory);
                            listHistory.Add(dequeueTbhistory);
                        }

                        lock (_icPrkmonitorContext)
                        {
                            if (!(listHistory.Count > 0))
                            {
                                return null;
                            }
                            str += $" List<Tbhistory> 中有{listHistory.Count} 条 \r\n";
                            _icPrkmonitorContext.Tbhistory.AddRange(listHistory);
                            var savecount = _icPrkmonitorContext.SaveChanges();
                            return str += $"代码执行时间：{DateTime.Now} 入库时间:{DateTime.Now} 成功添加数据库" + savecount + "条数据";
                        }
                    }
                }
            }
            catch (Exception e)
            {
                _logger.LogError(e.Message);
                return null;
            }
        }
    }

}
